package com.kuangstudy.controller.stream;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 飞哥
 * @Title: 学相伴出品
 * @Description: 我们有一个学习网站：https://www.kuangstudy.com
 * @date 2021/6/11 10:33
 */
@RestController
public class IMController {

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 初始化一个房间号
     *
     * @return
     */
    @GetMapping("/initmsg")
    public long initmessage() {
        long roomid = this.redisTemplate.opsForValue().increment("roomid");
        return roomid;
    }


    /**
     * 添加消息
     *
     * @param roomid
     * @param username
     * @param content
     * @return
     */
    @GetMapping("/addmsg")
    public RecordId addmessage(String roomid, String username, String content) {
        // 房间号
        String roomkey = "room:" + roomid;
        Map<String, String> map = new HashMap<>();
        map.put(username, content);
        RecordId recordId = this.redisTemplate.opsForStream().add(roomkey, map);
        return recordId;
    }

    /**
     * 监听消息
     * @param roomid
     * @return
     */
    @GetMapping("/readmsg")
    public List<MapRecord<String, String, String>> readmessage(String roomid) {
        // 房间号
        String roomkey = "room:" + roomid;
//        ReadOffset.latest() : 消息从最后开始读取 相当于 $
//        ReadOffset.from("ID"); 消息从指定的ID读取
        // 消息从那个位置开始读取
        StreamOffset<String> stringStreamOffset = StreamOffset.create(roomkey, ReadOffset.latest());
        //block阻塞10分钟进行读取，
        StreamReadOptions streamReadOptions = StreamReadOptions.empty().block(Duration.ofMinutes(10));
        return this.redisTemplate.opsForStream().read(streamReadOptions, stringStreamOffset);
    }

}
